package com.bawei.gmall.realtime.app

import com.alibaba.fastjson.JSON
import com.bawei.gmall.common.GmallConstants
import com.bawei.gmall.realtime.bean.{CouponAlertInfo, EventInfo}
import com.bawei.gmall.realtime.util.{EsUtil, KafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.InputDStream
import java.util

import scala.util.control.Breaks._

object AlertApp {

  /**
    * 需求：同一设备5分钟内三次及以上用不同账号登录并领取优惠劵，并且在登录到领劵过程中没有浏览商品,
    * 达到以上要求则产生一条预警日志。同一设备，每分钟只记录一次预警。
    *
    * 拆 ：
    * 同一设备 ： groupby mid
    * 5分钟内 ： 窗口大小（window）(范围)           窗口大小 ： 范围， 步长：统计频率
    * 用不同账号登录并领取优惠劵，没有浏览商品： 多变少， map变化结果（预警数据结构） filter把不符合条件过滤掉
    * 每分钟只记录一次预警 ： 开窗之后操作；
    *
    * @param args
    */
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("dau_app").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))
    // 1. 消费kafka中的数据
    val inputDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT, ssc)
    // 2. 转换结构
    val eventInfoDStream = inputDStream.map {
      record => {
        val info = JSON.parseObject(record.value(), classOf[EventInfo])
        info
      }
    }
    // 3. 5分钟内 ： 窗口大小（window）(范围)           窗口大小 ： 范围， 步长：统计频率
    val windowDStream = eventInfoDStream.window(Seconds(300), Seconds(5))
    // 4. 同一设备 ： groupby mid
    val groupByKeyMidDStream = windowDStream.map(eventInfo => (eventInfo.mid, eventInfo)).groupByKey()
    // 5. 用不同账号登录并领取优惠劵，没有浏览商品： 多变少， map变化结果（预警数据结构） filter把不符合条件过滤掉
    val checkDStream = groupByKeyMidDStream.map {
      case (mid, eventInfoItr) => {
        // 声明容器
        val couponUidSet = new util.HashSet[String]()
        val itemsSet = new util.HashSet[String]()
        val eventList = new util.ArrayList[String]()
        // 声明一个浏览商品的标识
        var isClickItem = false
        // 判断用户是否有浏览商品的行为
        breakable {
          for (elem <- eventInfoItr) {
            eventList.add(elem.evid)
            if (elem.evid == "coupon") {
              // 点击了购物券需要获取到当前的登录用户
              couponUidSet.add(elem.uid)
              itemsSet.add(elem.itemid)
            }
            if (elem.evid == "clickItem") {
              isClickItem = true
              break()
            }
          }
        }
        // 需求：同一设备5分钟内三次及以上用不同账号登录并领取优惠劵，并且在登录到领劵过程中没有浏览商品,
        // 达到以上要求则产生一条预警日志。同一设备，每分钟只记录一次预警。
        (couponUidSet.size() >= 3 && !isClickItem, CouponAlertInfo(mid, couponUidSet, itemsSet, eventList, System.currentTimeMillis()))
      }
    }
    // 6. 过滤符合条件的数据
    //    checkDStream.foreachRDD {
    //      rdd => println(rdd.collect().mkString("\n"))
    //    }
    val couponAlertInfo = checkDStream.filter(_._1).map(_._2)
    // 7. 保存到ES
    couponAlertInfo.foreachRDD {
      rdd => {
        rdd.foreachPartition {
          couponAlertInfoItr => {
            val list = couponAlertInfoItr.toList
            // 提取主键mid + 分钟组合为新的主键，在利用组合主键去重
            val tuples = list.map(couponAlertInfo => (couponAlertInfo.mid + "_" + couponAlertInfo.ts / 1000 / 60, couponAlertInfo))
            // 批量保存
            EsUtil.indexBulk(GmallConstants.ES_INDEX_ALERT,tuples)
          }
        }
      }
    }

    ssc.start()
    ssc.awaitTermination()
  }
}
